package com.rpc.client;

import com.rpc.transport.RequestData;
import com.rpc.transport.ResponseData;
import io.netty.channel.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

/**
 * @author yanzhuoleo
 * @version V1.0
 * @Description: 客户端连接中心
 * @date 2018-01-16-下午6:32
 */
public class ClientHandler extends SimpleChannelInboundHandler<ResponseData> {

    private Channel channel;
    private ConcurrentHashMap<String, RpcFuture> rpcMap         = new ConcurrentHashMap<>();
    private CountDownLatch                       countDownLatch = new CountDownLatch(1);
    private Logger                               logger         = LoggerFactory.getLogger(
            ClientHandler.class);

    /**
     * 接受服务端发来的消息
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ResponseData msg) {

        String    requestId = msg.getRequestId();
        RpcFuture rpcFuture = rpcMap.get(requestId);
        rpcMap.remove(requestId);
        if (rpcFuture != null && msg!=null) {
            rpcFuture.setResponseData(msg);
        }
    }

    /**
     * 与服务器建立连接
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        /**
         * TODO:这里为什么channel会不为空
         */
        if (channel != null) {
            channel.close();
        }
        channel = ctx.channel();
        System.out.println("channelActive");
    }

    /**
     * 与服务器断开连接
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        System.out.println("channelInactive");
    }

    /**
     * 异常
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        //关闭管道
        ctx.channel().close();
        //打印异常信息
        cause.printStackTrace();
    }

    public RpcFuture sendRequest(RequestData requestData) {

        RpcFuture rpcFuture = new RpcFuture();
        rpcFuture.setRequestData(requestData);
        rpcMap.put(requestData.getRequestId(), rpcFuture);
        channel.writeAndFlush(requestData).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) {
                countDownLatch.countDown();
            }
        });
        try {
            countDownLatch.await();
        } catch (InterruptedException ex) {
            //TODO:
            logger.warn(ex.getMessage());
        }
        return rpcFuture;

    }

}